package com.example.matchingengine.service;

import com.example.matchingengine.config.TopicConfig;
import com.example.matchingengine.model.Order;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 动态订单消费者服务
 * 支持为每个币对创建独立的消费者
 */
@Slf4j
@Service
public class DynamicOrderConsumerService {

    @Autowired
    private MeterRegistry meterRegistry;

    @Autowired
    private TopicConfig topicConfig;
    
    @Autowired
    private MatchingEngineService matchingEngineService;
    
    @Autowired
    private ObjectMapper objectMapper;
    
    @Value("${rocketmq.name-server:localhost:9876}")
    private String nameServer;
    
    @Value("#{'${matching-engine.engine.symbols}'.split(',')}")
    private List<String> symbols;

    // 存储每个币对的消费者
    private final Map<String, DefaultMQPushConsumer> orderConsumers = new ConcurrentHashMap<>();
    private final Map<String, DefaultMQPushConsumer> orderStatusConsumers = new ConcurrentHashMap<>();
    private final Map<String, DefaultMQPushConsumer> tradeConsumers = new ConcurrentHashMap<>();
    private final Map<String, DefaultMQPushConsumer> cancelOrderConsumers = new ConcurrentHashMap<>();

    // 计数器
    private Counter orderConsumeCounter;
    private Counter tradeConsumeCounter;

    @PostConstruct
    public void init() {
        // 初始化监控指标
        orderConsumeCounter = Counter.builder("matching.engine.order.consume.count")
                .description("消费订单数量")
                .register(meterRegistry);

        tradeConsumeCounter = Counter.builder("matching.engine.trade.consume.count")
                .description("消费成交事件数量")
                .register(meterRegistry);
        
        // 为所有配置的交易对创建消费者
        for (String symbol : symbols) {
            try {
                createConsumersForSymbol(symbol);
                log.info("已为币对 {} 创建所有消费者", symbol);
            } catch (Exception e) {
                log.error("为币对 {} 创建消费者失败: {}", symbol, e.getMessage(), e);
            }
        }
    }
    
    /**
     * 为指定币对创建消费者
     * @param symbol 币对符号
     */
    public void createConsumersForSymbol(String symbol) {
        try {
            createOrderConsumer(symbol);
            createOrderStatusConsumer(symbol);
            createTradeConsumer(symbol);
            createCancelOrderConsumer(symbol);
            log.info("已为币对 {} 创建所有消费者", symbol);
        } catch (Exception e) {
            log.error("为币对 {} 创建消费者失败: {}", symbol, e.getMessage(), e);
            throw new RuntimeException("创建消费者失败", e);
        }
    }
    
    /**
     * 创建订单消费者
     */
    private void createOrderConsumer(String symbol) throws MQClientException {
        String topic = topicConfig.getOrderTopic(symbol);
        String consumerGroup = topicConfig.getConsumerGroup(symbol, "order");
        
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe(topic, "*");
        
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> messages,
                    ConsumeConcurrentlyContext context) {
                
                for (MessageExt message : messages) {
                    try {
                        String messageBody = new String(message.getBody());
                        log.info("收到订单消息 from topic {}: {}", topic, messageBody);
                        
                        // 解析订单消息
                        Order order = objectMapper.readValue(messageBody, Order.class);
                        
                        // 调用撮合引擎处理订单
                        matchingEngineService.processOrder(order);
                        
                        orderConsumeCounter.increment();
                        log.info("订单处理完成: {}", order.getId());
                        
                    } catch (Exception e) {
                        log.error("处理订单消息失败: {}", e.getMessage(), e);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        
        consumer.start();
        orderConsumers.put(symbol, consumer);
        log.info("订单消费者已启动 - Topic: {}, ConsumerGroup: {}", topic, consumerGroup);
    }
    
    /**
     * 创建订单状态消费者
     */
    private void createOrderStatusConsumer(String symbol) throws MQClientException {
        String topic = topicConfig.getOrderStatusTopic(symbol);
        String consumerGroup = topicConfig.getConsumerGroup(symbol, "order-status");
        
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe(topic, "*");
        
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> messages,
                    ConsumeConcurrentlyContext context) {
                
                for (MessageExt message : messages) {
                    try {
                        log.info("收到订单状态消息 from topic {}: {}", topic, new String(message.getBody()));
                    } catch (Exception e) {
                        log.error("处理订单状态消息失败: {}", e.getMessage(), e);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        
        consumer.start();
        orderStatusConsumers.put(symbol, consumer);
        log.info("订单状态消费者已启动 - Topic: {}, ConsumerGroup: {}", topic, consumerGroup);
    }
    
    /**
     * 创建成交事件消费者
     */
    private void createTradeConsumer(String symbol) throws MQClientException {
        String topic = topicConfig.getTradeTopic(symbol);
        String consumerGroup = topicConfig.getConsumerGroup(symbol, "trade");
        
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe(topic, "*");
        
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> messages,
                    ConsumeConcurrentlyContext context) {
                
                for (MessageExt message : messages) {
                    try {
                        log.info("收到成交事件消息 from topic {}: {}", topic, new String(message.getBody()));
                        tradeConsumeCounter.increment();
                    } catch (Exception e) {
                        log.error("处理成交事件消息失败: {}", e.getMessage(), e);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        
        consumer.start();
        tradeConsumers.put(symbol, consumer);
        log.info("成交事件消费者已启动 - Topic: {}, ConsumerGroup: {}", topic, consumerGroup);
    }
    
    /**
     * 创建取消订单消费者
     */
    private void createCancelOrderConsumer(String symbol) throws MQClientException {
        String topic = topicConfig.getCancelOrderTopic(symbol);
        String consumerGroup = topicConfig.getConsumerGroup(symbol, "cancel-order");
        
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe(topic, "*");
        
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> messages,
                    ConsumeConcurrentlyContext context) {
                
                for (MessageExt message : messages) {
                    try {
                        String messageBody = new String(message.getBody());
                        log.info("收到取消订单消息 from topic {}: {}", topic, messageBody);
                        
                        // 解析订单消息
                        Order order = objectMapper.readValue(messageBody, Order.class);
                        
                        // 调用撮合引擎的取消订单方法
                        matchingEngineService.cancelOrder(order.getId(), order.getSymbol());
                        
                        log.info("取消订单处理完成: {}", order.getId());

                    } catch (Exception e) {
                        log.error("处理取消订单消息失败: {}", e.getMessage(), e);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        
        consumer.start();
        cancelOrderConsumers.put(symbol, consumer);
        log.info("取消订单消费者已启动 - Topic: {}, ConsumerGroup: {}", topic, consumerGroup);
    }
    
    /**
     * 停止指定币对的所有消费者
     */
    public void stopConsumersForSymbol(String symbol) {
        stopConsumer(orderConsumers.remove(symbol), "订单消费者");
        stopConsumer(orderStatusConsumers.remove(symbol), "订单状态消费者");
        stopConsumer(tradeConsumers.remove(symbol), "成交事件消费者");
        stopConsumer(cancelOrderConsumers.remove(symbol), "取消订单消费者");
        log.info("已停止币对 {} 的所有消费者", symbol);
    }
    
    private void stopConsumer(DefaultMQPushConsumer consumer, String consumerType) {
        if (consumer != null) {
            try {
                consumer.shutdown();
                log.info("{}已停止", consumerType);
            } catch (Exception e) {
                log.error("停止{}失败: {}", consumerType, e.getMessage(), e);
            }
        }
    }
    
    @PreDestroy
    public void destroy() {
        // 停止所有消费者
        orderConsumers.keySet().forEach(this::stopConsumersForSymbol);
        log.info("所有消费者已停止");
    }
}